package com.mng.transform;

import com.mng.dto.SensorReading;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;


public class StreamWordCount {
    public static void main(String[] args) throws Exception {
        //创建运行时流处理环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //设置并行度
        env.setParallelism(4);
        DataStream<String> stringDataStream = env.readTextFile("src/main/resources/wc.txt");
        //map string -->com.mng.dto.SensorReading
        DataStream<SensorReading> sensorReadingDataStream = stringDataStream.map(new MapFunction<String, SensorReading>() {
            @Override
            public SensorReading map(String s) throws Exception {
                String[] strings = s.split(",");
                String id = strings[0];
                Long timestamp = Long.parseLong(strings[1]);
                Double temp = Double.parseDouble(strings[2]);
                return new SensorReading(id,timestamp,temp);
            }
        });
        //flatmap String --> String.split
        DataStream<String> stringDataStreamWithFlat = stringDataStream.flatMap(new FlatMapFunction<String, String>() {
            @Override
            public void flatMap(String s, Collector<String> collector) throws Exception {
                String[] strings = s.split(",");
                for(String str : strings){
                    collector.collect(str);
                }
            }
        });
        //filter  sensorReading.id start with sensor_1
        DataStream<SensorReading> dataStreamWithFilter = sensorReadingDataStream.filter(new FilterFunction<SensorReading>() {
            @Override
            public boolean filter(SensorReading sensorReading) throws Exception {
                String id = sensorReading.getId();
                return id.startsWith("sensor_1");
            }
        });
        sensorReadingDataStream.print("map");
        stringDataStreamWithFlat.print("flatmap");
        dataStreamWithFilter.print("filter");
        env.execute();
    }
    
}
